package com.opages.transaction.message.service.impl;

import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.github.pagehelper.util.StringUtil;
import com.opages.transaction.message.common.exception.MessageException;
import com.opages.transaction.message.config.MessageProperties;
import com.opages.transaction.message.dao.TransactionMessageDao;
import com.opages.transaction.message.enums.MessageStatusEnum;
import com.opages.transaction.message.enums.PublicEnum;
import com.opages.transaction.message.model.TransactionMessage;
import com.opages.transaction.message.service.TransactionMessageService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author daizhong.liu
 * @create 2021-11-01 19:38
 **/
@Service
public class TransactionMessageServiceImpl implements TransactionMessageService {
    private static final Log log = LogFactory.getLog(TransactionMessageServiceImpl.class);
    @Autowired
    private MessageProperties messageProperties;

    @Autowired
    TransactionMessageDao transactionMessageDao;

    @Override
    public int saveMessageWaitingConfirm(TransactionMessage message) {
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "保存的消息为空");
        }
        if (StringUtil.isEmpty(message.getConsumerQueue())) {
            throw new MessageException(MessageException.MESSAGE_CONSUMER_QUEUE_IS_NULL, "消息的消费队列不能为空 ");
        }
        message.setUpdateDatetime(new Date());
        message.setStatus(MessageStatusEnum.WAITING_CONFIRM.value());
        message.setAreadlyDead(PublicEnum.NO.value());
        message.setMessageSendTimes(0);
        message.initBaseData();
        return transactionMessageDao.insert(message);
    }

    @Override
    public void updateMsgByConfirmAndSendMessage(String uuid) {
        final TransactionMessage message = getMessageByMessageId(uuid);
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "根据消息uuid查找的消息为空");
        }
        TransactionMessage updateMsg = new TransactionMessage();
        updateMsg.setUuid(uuid);
        updateMsg.setStatus(MessageStatusEnum.SENDING.value());
        updateMsg.setUpdateDatetime(new Date());
        //减少更新的字段数
        transactionMessageDao.updateByUuid(updateMsg);
        //AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
    }

    @Override
    public int saveAndSendMessage(TransactionMessage message) {
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "保存的消息为空");
        }
        if (StringUtil.isEmpty(message.getConsumerQueue())) {
            throw new MessageException(MessageException.MESSAGE_CONSUMER_QUEUE_IS_NULL, "消息的消费队列不能为空 ");
        }
        message.setStatus(MessageStatusEnum.SENDING.value());
        message.setAreadlyDead(PublicEnum.NO.value());
        message.setMessageSendTimes(0);
        message.initBaseData();
        int result = transactionMessageDao.insert(message);
//        AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
        return result;
    }

    @Override
    public void directSendMessage(TransactionMessage message) {
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "保存的消息为空");
        }
        if (StringUtil.isEmpty(message.getConsumerQueue())) {
            throw new MessageException(MessageException.MESSAGE_CONSUMER_QUEUE_IS_NULL, "消息的消费队列不能为空 ");
        }
//        AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
    }

    @Override
    public void updateMsgByReSendMessage(TransactionMessage message) {
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "保存的消息为空");
        }
        if (StringUtil.isEmpty(message.getConsumerQueue())) {
            throw new MessageException(MessageException.MESSAGE_CONSUMER_QUEUE_IS_NULL, "消息的消费队列不能为空 ");
        }
        message.addSendTimes();

        //减少更新字段数
        TransactionMessage updateMsg = new TransactionMessage();
        updateMsg.setUuid(message.getUuid());
        updateMsg.setMessageSendTimes(message.getMessageSendTimes());
        updateMsg.setUpdateDatetime(new Date());
        transactionMessageDao.updateByUuid(updateMsg);

//        AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
    }

    @Override
    public void updateMsgReSendMessageByMessageId(String uuid) {
        final TransactionMessage message = getMessageByMessageId(uuid);
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "根据消息id查找的消息为空");
        }

        if (message.getMessageSendTimes() >= messageProperties.getSender().getMaxTimes()) {
            message.setAreadlyDead(PublicEnum.YES.value());
        }
        message.addSendTimes();

        //减少更新字段数
        TransactionMessage updateMsg = new TransactionMessage();
        updateMsg.setUuid(uuid);
        updateMsg.setMessageSendTimes(message.getMessageSendTimes());
        updateMsg.setUpdateDatetime(new Date());
        transactionMessageDao.updateByUuid(updateMsg);

//        AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
    }

    @Override
    public void updateMessageToAreadlyDead(String uuid) {
        TransactionMessage message = getMessageByMessageId(uuid);
        if (message == null) {
            throw new MessageException(MessageException.SAVA_MESSAGE_IS_NULL, "根据消息uuid查找的消息为空");
        }
        //减少更新字段数
        TransactionMessage updateMsg = new TransactionMessage();
        updateMsg.setUuid(uuid);
        updateMsg.setAreadlyDead(PublicEnum.YES.value());
        updateMsg.setUpdateDatetime(new Date());
        transactionMessageDao.updateByUuid(updateMsg);
    }

    @Override
    public TransactionMessage getMessageByMessageId(String uuid) {
        TransactionMessage entity = new TransactionMessage();
        entity.setUuid(uuid);
        return transactionMessageDao.get(entity);
    }

    @Override
    public void deleteMessageByMessageId(String uuid) {
        TransactionMessage entity = new TransactionMessage();
        entity.setUuid(uuid);
        transactionMessageDao.delete(entity);
    }

    @Override
    public void updateMsgByReSendAllDeadMessageByQueueName(String queueName, int batchSize) {
        log.info("==>updateMsgByReSendAllDeadMessageByQueueName");

        int numPerPage = 1000;
        if (batchSize > 0 && batchSize < 100){
            numPerPage = 100;
        } else if (batchSize > 100 && batchSize < 5000){
            numPerPage = batchSize;
        } else if (batchSize > 5000){
            numPerPage = 5000;
        } else {
            numPerPage = 1000;
        }

        int pageNum = 1;
        Map<String, TransactionMessage> messageMap = new HashMap<String, TransactionMessage>();
        long pageCount = 1;
        TransactionMessage entity = new TransactionMessage();
        List<TransactionMessage> recordList =  null;
        entity.setConsumerQueue(queueName);
        entity.setAreadlyDead(PublicEnum.YES.value());
        PageHelper.startPage(pageNum,numPerPage," insert_datetime ASC ");
        List<TransactionMessage> list = transactionMessageDao.findList(entity);
        PageInfo<TransactionMessage> page = new PageInfo<>(list);
        recordList = page.getList();
//        entity.setOrderBy(" insert_datetime ASC ");
//        Page page = findPage(pageNum, numPerPage, entity);
        if (list == null || list.isEmpty()) {
            log.info("==>recordList is empty");
            return;
        }
        pageCount = page.getTotal();
        for (final TransactionMessage message : recordList) {
            messageMap.put(message.getUuid(), message);
        }
        for (pageNum = 2; pageNum <= pageCount; pageNum++) {
//            page = findPage(pageNum, numPerPage, entity);
            PageHelper.startPage(pageNum,numPerPage," insert_datetime ASC ");
            page = new PageInfo<>(list);
            recordList = page.getList();
            if (recordList == null || recordList.isEmpty()) {
                break;
            }
            for (final Object obj : recordList) {
                final TransactionMessage message = (TransactionMessage) obj;
                messageMap.put(message.getUuid(), message);
            }
        }

        recordList = null;
        page = null;

        for (Map.Entry<String, TransactionMessage> entry : messageMap.entrySet()) {
            final TransactionMessage message = entry.getValue();

            message.setUpdateDatetime(new Date());
            message.setMessageSendTimes(message.getMessageSendTimes() + 1);
            transactionMessageDao.update(message);
//            AliMqUtils.sendMsg(topic,message.getConsumerQueue(), message,message.getConsumerQueue());
        }
    }

    @Override
    public List<TransactionMessage> findList(TransactionMessage message) {
        return transactionMessageDao.findList(message);
    }
}
